package com.zisu.secs.secs.hsmsss;

import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;

import com.zisu.secs.secs.AbstractSecsInnerEngine;
import com.zisu.secs.secs.ReadOnlyTimeProperty;
import com.zisu.secs.secs.SecsException;
import com.zisu.secs.secs.SecsSendMessageException;
import com.zisu.secs.secs.SecsWaitReplyMessageException;
import com.zisu.secs.secs.secs2.Secs2BuildException;
import com.zisu.secs.secs.secs2.Secs2ByteBuffersBuilder;

public class HsmsSsSendReplyManager extends AbstractSecsInnerEngine {
	
	private final Collection<Pack> packs = new ArrayList<>();
	
	private final AbstractHsmsSsCommunicator parent;
	
	public HsmsSsSendReplyManager(AbstractHsmsSsCommunicator parent) {
		super(parent);
		this.parent = parent;
	}
	
	public Optional<HsmsSsMessage> send(HsmsSsMessage msg)
			throws SecsSendMessageException, SecsWaitReplyMessageException, SecsException
			, InterruptedException {
		
		AsynchronousSocketChannel channel = parent.optionalChannel()
				.orElseThrow(() -> new HsmsSsNotConnectedException(msg));
		
		HsmsSsMessageType type = HsmsSsMessageType.get(msg);
		
		switch ( type ) {
			case SELECT_REQ:  //如果是SELECT_REQ 类型的，，因为没有break，那么SELECT_REQ请求都会用后面LINKTEST_REQ/DATA 方式来执行
			case LINKTEST_REQ: {
				//把这个，msg 包装成Pack 然后注册到全局 的 packs 里面，方便后续的监听（reply()函数）
				Pack p = entry(msg);
				try {
					send(channel, msg);
					/**
					 * 可能1
					 * 		com.zisu.secs.secs.hsmsss.HsmsSsTimeoutT6Exception: [FF FF|00 00|00 05|00 00 00 04]
					 * 		 at com.zisu.secs.secs.hsmsss.HsmsSsSendReplyManager.send(HsmsSsSendReplyManager.java:54)
					 *
					 * 		如果报这个错，那么就说明是EQ端出问题了（HOST给EQ发Select.req 指令 但是EQ端没有在t6的时间范围之内回复主机端Select.rsp指令）
					 *
					 * 可能2
					 * 		"HSMS requires the reject procedure for the receipt of a data message in the NOT SELECTED state"
					 * 		就是模拟器（EQ端） 一段时间后 会自动把HOST端给 断掉，这样HOST端 再发心跳给EQ的时候就会报错
					 * */
					try {
						//如果是SELECT_REQ 类的指令 也会到这里。（包装一个 会送命令 返回回去）
						return Optional.of(reply(p, parent.hsmsSsConfig().timeout().t6()));
					}
					catch ( TimeoutException e ) {
						throw new HsmsSsTimeoutT6Exception(msg, e);
					}
				}
				//finally语句总是要执行的！ 如果上面有return ；； try与finally里面都有return语句；这种情况忽略try中的return语句，所以最终返回的是finally语句中的返回结果
				finally {
					remove(p);
				}
				/* break; */
			}
			//与其说DATA类型 倒不如说 是 需要回复 命令（指令这个代码块）
			case DATA: {
				if ( msg.wbit() ) {
					Pack p = entry(msg);
					try {
						send(channel, msg);
						//其实 还是 HOST发给 EQ 的Device id 错了， device id 第一个字节和第二个字节 不能不一样
						//Thread.sleep(200);
						try {
							return Optional.of(reply(p, parent.hsmsSsConfig().timeout().t3()));
						}
						catch ( TimeoutException e ) {
							throw new HsmsSsTimeoutT3Exception(msg, e);
						}
					}
					finally {
						remove(p);
					}
				} else {
					send(channel, msg);
					return Optional.empty();
				}
				/* break */
			}
			default: {
				//除了LINKTEST_REQ 和 DATA 的 数据类别 都会到这里来。
				send(channel, msg);
				return Optional.empty();
			}
		}
	}
	
	private static final long MAX_BUFFER_SIZE = 256L * 256L;
	
	protected long prototypeMaxBufferSize() {
		return MAX_BUFFER_SIZE;
	}
	
	public void send(AsynchronousSocketChannel channel, HsmsSsMessage msg)
			throws SecsSendMessageException, SecsException
			, InterruptedException {
		
		synchronized ( channel ) {
			
			try {
				Secs2ByteBuffersBuilder bb = Secs2ByteBuffersBuilder.build(1024, msg.secs2());
				
				long len = bb.size() + 10L;
				
				if ((len > 0x00000000FFFFFFFFL) || (len < 10L)) {
					throw new HsmsSsTooBigSendMessageException(msg);
				}
				
				notifyTrySendMessagePassThrough(msg);
				
				long bufferSize = len + 4L;
				
				if ( bufferSize > prototypeMaxBufferSize() ) {
					
					{
						ByteBuffer buffer = ByteBuffer.allocate(14);
						
						buffer.put((byte)(len >> 24));
						buffer.put((byte)(len >> 16));
						buffer.put((byte)(len >>  8));
						buffer.put((byte)(len      ));
						buffer.put(msg.header10Bytes());
						
						((Buffer)buffer).flip();
						
						send(channel, buffer);
					}
					
					for ( ByteBuffer buffer : bb.getByteBuffers() ) {
						send(channel, buffer);
					}
					
				} else {
					
					ByteBuffer buffer = ByteBuffer.allocate((int)bufferSize);
					
					buffer.put((byte)(len >> 24));
					buffer.put((byte)(len >> 16));
					buffer.put((byte)(len >>  8));
					buffer.put((byte)(len      ));
					buffer.put(msg.header10Bytes());
					
					for ( ByteBuffer bf : bb.getByteBuffers() ) {
						buffer.put(bf);
					}
					
					((Buffer)buffer).flip();
					
					send(channel, buffer);
				}
				
				notifySendedMessagePassThrough(msg);
				notifyLog("Sended HsmsSs-Message", msg);
			}
			catch ( ExecutionException e ) {
				
				Throwable t = e.getCause();
				
				if ( t instanceof RuntimeException ) {
					throw (RuntimeException)t;
				}
				
				throw new HsmsSsSendMessageException(msg, e);
			}
			catch ( Secs2BuildException | HsmsSsDetectTerminateException e ) {
				throw new HsmsSsSendMessageException(msg, e);
			}
		}
	}
	
	private void send(AsynchronousSocketChannel channel, ByteBuffer buffer)
			throws ExecutionException, HsmsSsDetectTerminateException, InterruptedException {
		
		while ( buffer.hasRemaining() ) {
			
			final Future<Integer> f = channel.write(buffer);
			
			try {
				int w = f.get().intValue();
				
				if ( w <= 0 ) {
					throw new HsmsSsDetectTerminateException();
				}
			}
			catch ( InterruptedException e ) {
				f.cancel(true);
				throw e;
			}
		}
	}

	/**
	 * reply 这个方法的是指 是 等回复 指令比如 我发117 就等待 118指令
	 * */
	private HsmsSsMessage reply(Pack p, ReadOnlyTimeProperty timeout)
			throws SecsWaitReplyMessageException, SecsException
			, TimeoutException, InterruptedException {

		final Callable<HsmsSsMessage> getMsgTask = () -> {
			
			try {
				synchronized ( packs ) {
					for ( ;; ) {
						//一直循环 去获取 EQ返回给我的 “回复”
						HsmsSsMessage m = p.replyMsg();
						if ( m != null ) {
							return m;
						}
						packs.wait();
					}
				}
			}
			catch ( InterruptedException ignroe ) {
			}
			
			return null;
		};
		
		final Callable<HsmsSsMessage> checkTerminateTask = () -> {
			
			try {
				synchronized ( packs ) {
					
					for ( ;; ) {
						
						if ( packs.isEmpty() ) {
							return null;
						}
						
						packs.wait();
					}
				}
			}
			catch ( InterruptedException ignore ) {
			}
			
			return null;
		};
		
		try {
			//提交给线程池 也就是异步的,这里的msg应该就是 getMsgTask 返回过来的 p.replyMsg();
			HsmsSsMessage msg = executeInvokeAny(getMsgTask, checkTerminateTask, timeout);
			
			if ( msg == null ) {
				throw new HsmsSsDetectTerminateException();
			}

			//REJECT_REQ 一定是外部put 进去的
			if ( HsmsSsMessageType.get(msg) == HsmsSsMessageType.REJECT_REQ ) {
				throw new HsmsSsRejectException(msg);
			}
			
			return msg;
		}
		catch ( ExecutionException e ) {
			
			Throwable t = e.getCause();
			
			if ( t instanceof RuntimeException ) {
				throw (RuntimeException)t;
			}
			
			throw new SecsException(e);
		}
	}


	public void clear() {
		synchronized ( packs ) {
			packs.clear();
			packs.notifyAll();
		}
	}
	
	private Pack entry(HsmsSsMessage msg) {
		synchronized ( packs ) {
			Pack p = new Pack(msg);
			packs.add(p);
			return p;
		}
	}
	
	private void remove(Pack p) {
		synchronized ( packs ) {
			packs.remove(p);
		}
	}

	// 依据msg的 systemcode对其分类然后塞入对应的队列，，然后唤醒所有阻塞的队列（队列有数据的话会正常执行，队列中无数据就会继续阻塞住）
	public Optional<HsmsSsMessage> put(HsmsSsMessage msg) {
		
		final Integer key = msg.systemBytesKey();
		
		synchronized ( packs ) {
			
			for ( Pack p : packs ) {
				
				if ( p.key().equals(key) ) {
					//p.put 就是 set replyMsg  并且唤醒 之前等待 这个msg.replyMsg的线程 去拿到这个 replyMsg
					p.put(msg);
					packs.notifyAll();
					return Optional.empty();
				}
			}
			//Optional.of() 为msg 创建一个容器
			return Optional.of(msg);
		}
	}
	
	private class Pack {
		
		private final HsmsSsMessage primary;
		private final Integer key;
		private HsmsSsMessage reply;
		
		public Pack(HsmsSsMessage primaryMsg) {
			this.primary = primaryMsg;
			this.key = primary.systemBytesKey();
			this.reply = null;
		}
		
		public Integer key() {
			return key;
		}
		
		public void put(HsmsSsMessage replyMsg) {
			synchronized ( this ) {
				this.reply = replyMsg;
			}
		}
		
		public HsmsSsMessage replyMsg() {
			synchronized ( this ) {
				return reply;
			}
		}
		
		@Override
		public int hashCode() {
			return key.hashCode();
		}
		
		@Override
		public boolean equals(Object o) {
			if ((o != null) && (o instanceof Pack)) {
				return ((Pack)o).key.equals(key);
			} else {
				return false;
			}
		}
	}

}
